package com.cryann.mqttserver.mqttProcess;

import com.cryann.mqttserver.config.AsyScheduler;
import com.cryann.mqttserver.exception.MqttException;
import com.cryann.mqttserver.modle.AccessorStore;
import com.cryann.mqttserver.service.MqttAccessor;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author cRyann
 * @date 2018/12/14 0014 11:31
 * @description:
 */
@Component
@Slf4j
@ProcessType(MqttMessageType.CONNECT)
public class Connect implements BasicProcess<MqttConnectMessage> {

    @Autowired
    MqttAccessor mqttAccessor;
    @Autowired
    AsyScheduler asyScheduler;

    @Override
    public void process(Channel ctx, MqttConnectMessage message) {
        MqttConnectVariableHeader header = message.variableHeader();
        MqttConnectPayload payload = message.payload();
        //协议检查
        //协议名非法
//        if (!"MQTT".equals(message.variableHeader().name())) {
//            log.warn("Channel from :[{}] 协议名非法", ctx.remoteAddress());
//            ctx.closeFuture();
//            return;
//        }
        //3.1.2.2 协议级别
        if ( 4 != header.version() ) {
            log.warn("Channel from :[{}] 协议级别非法", ctx.remoteAddress());
            MqttConnAckMessage msg = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 2),
                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false),
                    null
            );
            ctx.writeAndFlush(msg);
            ctx.closeFuture();
            return;
        }

        // netty 默认类不判断保留位(byte0 后4位)
        //处理protocol name和protocol version, 如果返回码!=0，sessionPresent必为0，协议P24,P32
        if ( !("MQTT").equals(header.name()) ||
                header.version() != 4 ) {
            MqttConnAckMessage mqttConnAckMessage = new MqttConnAckMessage(new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_LEAST_ONCE, false, 0)
                    , new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false));
            ctx.writeAndFlush(mqttConnAckMessage);
            ctx.close();//版本或协议名不匹配，则断开该客户端连接
            return;
        }
        MqttConnAckMessage mqttConnAckMessage = new MqttConnAckMessage(new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_LEAST_ONCE, false, 0)
                , new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, false));
        //TODO 需要增加配置文件读取
        if ( !"cryann".equals(payload.userName())
                || !"cryann".equals(new String(payload.passwordInBytes())) ) {
            log.debug("用户名或密码不正确");
            return;
        } else {
            log.info("用户:[{}] 连接成功 ", payload.userName());
        }
        // 增加连接用户 并 将成功登录的连接保存到channel连接池
        // TODO
        // [MQTT-3.1.2-4]	如果清理会话（CleanSession）标志被设置为0，服务端必须基于当前会话（使用客户端标识符识别）的状态恢复与客户端的通信。如果没有与这个客户端标识符关联的会话，服务端必须创建一个新的会话。在连接断开之后，当连接断开后，客户端和服务端必须保存会话信息。
        // [MQTT-3.1.2-5]	当清理会话标志为0的会话连接断开之后，服务端必须将之后的QoS 1和QoS 2级别的消息保存为会话状态的一部分，如果这些消息匹配断开连接时客户端的任何订阅。
        // [MQTT-3.1.2-6]	如果清理会话（CleanSession）标志被设置为1，客户端和服务端必须丢弃之前的任何会话并开始一个新的会话。会话仅持续和网络连接同样长的时间。与这个会话关联的状态数据不能被任何之后的会话重用。
        // [MQTT-3.1.2.7]	保留消息不是服务端会话状态的一部分，会话终止时不能删除保留消息。
        MqttConnectVariableHeader variableHeader = message.variableHeader();
//        MARK 更新AliveCheckTask时间  默认最小是10秒 避免服务器压力过大
//        更新逻辑 启动一个动态定时器，当新的keeplive值比keepalive值小，更新该定时器：以保证只使用个一线程来处理 AliveCheck
//        保持连接的值为零表示关闭保持连接功能。这意味着，服务端不需要因为客户端不活跃而断开连接。注意：不管保持连接的值是多少，任何时候，只要服务端认为客户端是不活跃或无响应的，可以断开客户端的连接。
        if ( variableHeader.keepAliveTimeSeconds() > 10
                && variableHeader.keepAliveTimeSeconds() < AccessorStore.CHECK_LIVE_INTERVAL_TIME ) {
            asyScheduler.UpdateAliveCheckTask(variableHeader.keepAliveTimeSeconds());
            log.debug("AliveCheckTask Time [{}] changed to [{}]", AccessorStore.CHECK_LIVE_INTERVAL_TIME, variableHeader.keepAliveTimeSeconds());
            AccessorStore.CHECK_LIVE_INTERVAL_TIME = variableHeader.keepAliveTimeSeconds();
        }
        if ( mqttAccessor.addAccessor(ctx, message) ) {
            ctx.writeAndFlush(mqttConnAckMessage);
        } else {
            throw new MqttException("[MQTT-3.1.0-1] [MQTT-3.1.0-2] ");
        }
    }
}
